package com.xiaok.netty.server.hander;

import com.xiaok.common.protobuf.Command;
import com.xiaok.common.protobuf.Message;
import com.xiaok.netty.server.components.ChannelRepository;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.concurrent.GlobalEventExecutor;
import javafx.css.CssMetaData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;


/**
 * 心跳检查处理类
 * @author xiaok
 * @create 2020--09-30 10:35
 */
@Component("serverHeartHandler")
@ChannelHandler.Sharable //ChannelHandler被注解为 @Sharable，全局只有一个handler实例，它会被多个Channel的Pipeline共享(没加sharable注解的话,一个连接对应一个ServerHeartHandler)
// ，会被多线程并发调用，
// 因此它不是线程安全的；如果存在跨ChannelHandler的实例级变量共享，需要特别注意，它可能不是线程安全的。
// 在整个ChannelPipeline执行过程中，可能会发生线程切换。此时，如果同一个对象在多个ChannelHandler中被共享，可能会被多线程并发操作
// 使用场景: 在你确认了你的 ChannelHandler 是线程安全的时才使用@Sharable ,比如 收集多个 Channel的统计信息
//Sharable注解的使用是有限制的，多个ChannelPipeline共用一个handler实例，所以该Handler要求无状态。不能使用或改变本身的私有变量，因为ChannelHandler是非线程安全的，使用私有变量会造成线程竞争而产生错误结果
//https://www.cnblogs.com/silyvin/p/9593368.html
public class ServerHeartHandler extends ChannelInboundHandlerAdapter {
    public Logger log = LoggerFactory.getLogger(this.getClass());

    private final AttributeKey<String> clientInfo = AttributeKey.valueOf("clientInfo");

    //定义一个channel组,管理所有的channel
    //GlobalEventExecutor.INSTANCE) 是全局的事件执行器，是一个单例
     private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    // 设置6秒检测chanel是否接受过心跳数据
    private static final int READ_WAIT_SECONDS = 6;

    // 定义客户端没有收到服务端的pong消息的最大次数
    private static final int MAX_UN_REC_PING_TIMES = 3;

    // 失败计数器, 未收到client端发送的 ping 请求
    private Map<String,Integer> unRecPingTimeMap = new ConcurrentHashMap<String, Integer>();

    private DateTimeFormatter dtf =  DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); //代替simpleDateFormat
    @Autowired
    @Qualifier("channelRepository")
    private ChannelRepository channelRepository;

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        String data = "客户端IP:+ "+channel.remoteAddress()+"上线了,上线时间: "+dtf.format(LocalDateTime.now()); //LocalDateTime 代替calendar
        channelGroup.writeAndFlush(createData("",Command.CommandType.PUSH_DATA,data));
        channelGroup.add(channel);
    }

    /**
     * 建立连接
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info(ctx.channel().remoteAddress() + " 上线了~");
    }

    /*空闲触发器 心跳基于空闲实现*/
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {

        if( evt instanceof IdleStateEvent){
            IdleStateEvent event = (IdleStateEvent) evt;
            String type = "";
            if (event.state() == IdleState.READER_IDLE) { // 读空闲
                type = "read idle";
            } else if (event.state() == IdleState.WRITER_IDLE) { //写空闲
                type = "write idle";
            } else if (event.state() == IdleState.ALL_IDLE) { //读写空闲
                type = "all idle";
            }
            log.info("触发空闲,type = "+type);
            Attribute<String> attr = ctx.channel().attr(clientInfo);
            String clientId = attr.get();
            log.info("clientId = "+clientId);
            Integer unRecPingTimes = unRecPingTimeMap.get(clientId);
            if( unRecPingTimes >= MAX_UN_REC_PING_TIMES){
                // 连续超过N次未收到client的ping消息，那么关闭该通道，等待client重连
                ctx.channel().close();
            }else{ //失败次数加1
                unRecPingTimeMap.put(clientId,++unRecPingTimes);
            }
            log.info("超时次数：" + unRecPingTimes);
        }else{// 调用父类的心跳触发逻辑
            super.userEventTriggered(ctx,evt);
        }
    }


    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        Message.MessageBase msgBase = (Message.MessageBase)msg; //获取传输的数据
        String clientId = msgBase.getClientId();
        Command.CommandType cmd = msgBase.getCmd();
        Channel ch = channelRepository.get("clientId");
        if( ch ==null ){
            Channel channel = ctx.channel();
            channelRepository.put(clientId,channel);

        }
        /* 认证处理 */
        String data ="";
        log.info("cmd: "+ cmd);
        if( cmd.equals(Command.CommandType.AUTH)){
            log.info("收到客户端Id是"+clientId+"的建立连接请求...");
            log.info("收到客户端的 Auth 数据: "+msgBase.getData());
            Attribute<String> attr = ctx.channel().attr(clientInfo);
            attr.set(clientId);
            unRecPingTimeMap.put(clientId,0);//将对应虚拟机的心跳次数置空
            channelRepository.put(clientId, ctx.channel());
            // 回写数据
            data = "客户端Id是"+clientId+"的建立连接请求success！";
            ctx.writeAndFlush(createData(clientId,Command.CommandType.AUTH_BACK,data));
        }else if( msgBase.getCmd().equals(Command.CommandType.PING)){ //开始心跳
            data = "服务器响应客户端Id是"+clientId+"的心跳包！";
            log.info("data: "+data);
            ctx.writeAndFlush(createData(clientId,Command.CommandType.PONG,data));
        }else{
            if(ch.isOpen()){
                //触发下一个handler,此 msg非心跳检查的
                log.info("触发下一个handler");
                ctx.fireChannelRead(msg);
            }
        }
        // 释放资源, 保存ByteBuf的消息
        ReferenceCountUtil.release(msg);
    }

    /**
     * 异常处理
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        Attribute<String> attr = ctx.channel().attr(clientInfo); //获取机器的信息
        String clientId = attr.get();
        log.error("客户端断开链接，Id是 " + clientId+"----错误详情是======"+cause.getMessage());
    }

    //表示channel 处于不活动状态, 提示 xx离线了
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.info(ctx.channel().remoteAddress() + " 离线了~");
    }

    /**
     * 创建 protobuf 的不同类型的数据
     * @param clientId
     * @param cmd
     * @param data
     * @return
     */
    private Message.MessageBase.Builder createData(String clientId, Command.CommandType cmd, String data){
        Message.MessageBase.Builder msg = Message.MessageBase.newBuilder();
        //封装数据
        msg.setClientId(clientId);
        msg.setCmd(cmd);
        msg.setData(data);
        return msg;

    }
}
